/*
 * @Description:
 * @Author: Dxcr
 * @Date: 2025-01-15 09:53:10
 * @LastEditTime: 2025-01-16 10:07:23
 * @LastEditors: Dxcr
 */
use rumqttc::{AsyncClient, Event, EventLoop, MqttOptions, Packet, QoS};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

#[derive(Clone)]
pub struct MqttClient {
    client: AsyncClient,
    event_loop: Option<Arc<Mutex<EventLoop>>>,
}

impl MqttClient {
    pub fn new(id: &str, broker: &str, port: u16, username: &str, password: &str) -> Self {
        let mut mqttoptions = MqttOptions::new(id, broker, port);
        mqttoptions
            .set_keep_alive(Duration::from_secs(10))
            .set_credentials(username, password);

        let (client, event_loop) = AsyncClient::new(mqttoptions, 100);

        println!(
            "MQTT 客户端初始化成功\n\
            --------------------------\n\
            Id:          {}\n\
            Broker:      {}\n\
            Port:        {}\n\
            Username:    {}\n\
            Keep Alive:  10s\n\
            --------------------------",
            id, broker, port, username
        );

        Self {
            client,
            event_loop: Some(Arc::new(Mutex::new(event_loop))),
        }
    }

    //订阅主题接受数据
    pub async fn subscribe<F>(&self, topic: &str, callback: F)
    where
        F: Fn(String, String) + Send + Sync + 'static,
    {
        // 检查 EventLoop 是否已初始化
        if let Some(event_loop) = self.event_loop.as_ref() {
            // 尝试订阅指定的主题
            if let Err(e) = self.client.subscribe(topic, rumqttc::QoS::AtMostOnce).await {
                eprintln!("订阅主题 [{}] 失败: {}", topic, e);
                return;
            }
            println!("成功订阅主题: [{}]", topic);

            // 将回调函数包装在 Arc 中以便共享
            let callback = Arc::new(callback);
            let event_loop = event_loop.clone();

            // 启动一个异步任务处理消息
            tokio::task::spawn(async move {
                let mut locked_event_loop = event_loop.lock().await; // 获取 Mutex 锁
                loop {
                    // 拉取 EventLoop 中的事件
                    match locked_event_loop.poll().await {
                        Ok(Event::Incoming(Packet::Publish(publish))) => {
                            // 处理接收到的消息
                            let topic = publish.topic.clone();
                            let payload = String::from_utf8_lossy(&publish.payload).to_string();

                            println!("接收到消息: 主题 [{}], 内容 [{}]", topic, payload);

                            // 调用用户提供的回调函数
                            let callback = Arc::clone(&callback);
                            callback(topic, payload);
                        }
                        Ok(_) => {
                            // 忽略其他事件
                        }
                        Err(e) => {
                            eprintln!("EventLoop 处理事件时出错: {}", e);
                            break;
                        }
                    }
                }
            });
        } else {
            eprintln!("错误: EventLoop 未初始化");
        }
    }

    //指定主题发布数据
    pub async fn publish(&self, topic: &str, payload: &str) {
        match self
            .client
            .publish(topic, QoS::AtMostOnce, false, payload.as_bytes())
            .await
        {
            Ok(_) => println!("成功向 [{}] 发送数据: {}", topic, payload),
            Err(e) => eprintln!("向 [{}] 发送数据失败: {}", topic, e),
        }
    }

    // 使用 tokio 定时器每 5 秒向指定主题发送一次数据
    pub async fn publish_interval(&self, topic: &str, payload: &str, interval: u64) {
        // 将 self 包装为 Arc<Mutex<Self>>，以便在 tokio::spawn 中安全使用
        let publisher = Arc::new(Mutex::new(self.clone()));
        let topic = topic.to_string();
        let payload = payload.to_string();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(interval));
            loop {
                interval.tick().await; // 等待指定的间隔时间
                let publisher = publisher.lock().await;
                publisher.publish(&topic, &payload).await;
            }
        });
    }

    // 使用 tokio 定时器每 5 秒向指定主题发送一次数据
    pub async fn publish_interval_callback<F>(&self, topic: &str, mut payload_generator: F, interval: u64)
    where
        F: FnMut() -> String + Send + 'static,
    {
        // 将 self 包装为 Arc<Mutex<Self>>，以便在 tokio::spawn 中安全使用
        let publisher = Arc::new(Mutex::new(self.clone()));
        let topic = topic.to_string();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(interval));
            loop {
                interval.tick().await; // 等待指定的间隔时间
                let payload = payload_generator(); // 调用回调函数生成新的 payload
                let publisher = publisher.lock().await;
                publisher.publish(&topic, &payload).await;
            }
        });
    }
}
